package phoneinfo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Phonecount {

	/**统计第一季度用户的手机上网流量，通话时间，短信数量
	 * @param args
	 */
	public static void main(String[] args)throws Exception {
		if (args.length!=2) {
			System.err.println("user path err!");
			System.exit(-1);
		}
		Job job=new Job(new Configuration(), "jobName");
		job.setJarByClass(Phonecount.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		job.setMapperClass(phmap.class);
		job.setReducerClass(phreduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(PhoneNum.class);
		job.waitForCompletion(true);
		
		

	}
	public static class phmap extends Mapper<LongWritable, Text,Text,PhoneNum>{
		protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,PhoneNum>.Context context) throws java.io.IOException ,InterruptedException {
			String[] lines = value.toString().split("\t");
			if (lines.length==4) {
				String pnum=lines[0];
				int GPRS = Integer.parseInt(lines[1].trim());
				int phtime = Integer.parseInt(lines[2].trim());
				int phmail = Integer.parseInt(lines[3].trim());
				PhoneNum phoneNum = new PhoneNum(GPRS,phtime,phmail);
				context.write(new Text(pnum),phoneNum);
			}
			
		};
	}
	//map -->shuffle处理后：<1653285485,{phoneNum1,phoneNum2....}>
	public static class phreduce extends Reducer<Text, PhoneNum, Text,PhoneNum> {
		
		protected void reduce(Text k2, java.lang.Iterable<PhoneNum> value, org.apache.hadoop.mapreduce.Reducer<Text,PhoneNum,Text,PhoneNum>.Context context) throws java.io.IOException ,InterruptedException {
			int GPRSsum=0;
			int phtimesum=0;
			int phmailsum=0;
			for (PhoneNum phoneNum : value) {
				GPRSsum+=phoneNum.getInternate();
				phtimesum+=phoneNum.getCalltime();
				phmailsum+=phoneNum.getMailcount();
				//分别对上网时间，通话时间，短信数求和
				//将求和后的总量作为参数值
			}
			PhoneNum phoneNum2 = new PhoneNum(GPRSsum,phtimesum,phmailsum);
			context.write(k2,phoneNum2);
		};
	}

}
